package cn.baker.delayqueue.task;


import cn.baker.delayqueue.config.DelayQueueProperties;
import cn.baker.delayqueue.entity.DelayQueueMessage;
import cn.baker.delayqueue.util.RedisDelayQueueUtil;
import cn.baker.delayqueue.mapper.DelayQueueMessageMapper;
import cn.hutool.core.collection.CollUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author yuanyu
 */
@Slf4j
public class ScanTask {

    private final RedisDelayQueueUtil redisDelayQueueUtil;
    private final DelayQueueMessageMapper delayQueueMessageMapper;
    private final DelayQueueProperties delayQueueProperties ;

    public ScanTask(RedisDelayQueueUtil redisDelayQueueUtil, DelayQueueMessageMapper delayQueueMessageMapper, DelayQueueProperties delayQueueProperties) {
        this.redisDelayQueueUtil = redisDelayQueueUtil;
        this.delayQueueMessageMapper = delayQueueMessageMapper;
        this.delayQueueProperties = delayQueueProperties;
    }

    /**
     * 定时任务
     */
    @Scheduled(fixedDelayString = "${delay.queue.taskScanFreq:1000}")
    @Transactional(rollbackFor = Exception.class)
    public void scan() throws JsonProcessingException {
        // 找出过期时间
        long time = System.currentTimeMillis() + delayQueueProperties.getScanMaxTime();
        log.debug("do scan,time={}", time);
        while (true) {
            List<DelayQueueMessage> delayQueueMessageList = delayQueueMessageMapper.scanMessage(time, 0, 1000);
            if (CollUtil.isEmpty(delayQueueMessageList)) {
                log.debug("scan finish");
                return;
            }
            for (DelayQueueMessage delayQueueMessage : delayQueueMessageList) {
                redisDelayQueueUtil.addDelayQueue(
                        false,
                        delayQueueMessage.getPayload(),
                        delayQueueMessage.getDelayInMillis() - System.currentTimeMillis(),
                        TimeUnit.MILLISECONDS,
                        delayQueueMessage.getQueueCode()
                );
            }
            delayQueueMessageMapper.deleteByIds(delayQueueMessageList.stream().map(DelayQueueMessage::getId).collect(Collectors.toList()));
        }
    }


}
